package com.edu.realtime.app.dws;

import java.time.Duration;
import java.util.concurrent.TimeUnit;

import org.apache.flink.api.common.eventtime.SerializableTimestampAssigner;
import org.apache.flink.api.common.eventtime.WatermarkStrategy;
import org.apache.flink.api.common.functions.ReduceFunction;
import org.apache.flink.api.common.state.MapState;
import org.apache.flink.api.common.state.MapStateDescriptor;
import org.apache.flink.api.common.state.StateTtlConfig;
import org.apache.flink.api.common.time.Time;
import org.apache.flink.configuration.Configuration;
import org.apache.flink.streaming.api.datastream.*;
import org.apache.flink.streaming.api.environment.StreamExecutionEnvironment;
import org.apache.flink.streaming.api.functions.KeyedProcessFunction;
import org.apache.flink.streaming.api.functions.windowing.AllWindowFunction;
import org.apache.flink.streaming.api.windowing.assigners.TumblingEventTimeWindows;
import org.apache.flink.streaming.api.windowing.windows.TimeWindow;
import org.apache.flink.util.Collector;

import com.alibaba.fastjson.JSONObject;
import com.edu.realtime.app.func.DimAsyncFunction;
import com.edu.realtime.bean.TestCourseUcScoreDuration;
import com.edu.realtime.util.DateFormatUtil;
import com.edu.realtime.util.MyClickhouseUtil;
import com.edu.realtime.util.MyKafkaUtil;

/**
 * Created on 2022/10/20.
 *
 * @author Topus
 * @desc
 */
public class DwsTestCourseUcScoreDurationWindow {
    public static void main(String[] args) throws Exception {
        /*
        当日课程粒度考试的相关统计：
            从dwd_test_exam把数据读出来，两次流关联，从dim_test_paper中读取COURSE_ID->courseId，拿到之后从dim_course_info中读取COURSE_NAME->courseName
            根据courseId分组keyBy
            mapState获取独立用户数
            聚合求总的score、duration_sec
         */
        StreamExecutionEnvironment env = StreamExecutionEnvironment.getExecutionEnvironment();
        env.setParallelism(4);

        String topic = "dwd_test_exam";
        String groupId = "dws_test_course_uc_score_duration_window_group";
        DataStreamSource<String> source = env.addSource(MyKafkaUtil.getKafkaConsumer(topic, groupId));
        SingleOutputStreamOperator<JSONObject> mapDS = source.map(JSONObject::parseObject);
        // 获取test_paper中的courseId
        SingleOutputStreamOperator<JSONObject> paperDimDS =
            AsyncDataStream.unorderedWait(mapDS, new DimAsyncFunction<JSONObject>("dim_test_paper") {
                @Override
                public void join(JSONObject input, JSONObject dimInfo) {
                    input.put("course_id", dimInfo.getString("COURSE_ID"));
                }

                @Override
                public String getKey(JSONObject obj) {
                    return obj.getString("paper_id");
                }
            }, 120, TimeUnit.SECONDS);

        // 设置水位线
        SingleOutputStreamOperator<JSONObject> withWatermarkDS = paperDimDS
            .assignTimestampsAndWatermarks(WatermarkStrategy.<JSONObject>forBoundedOutOfOrderness(Duration.ofSeconds(3))
                .withTimestampAssigner(new SerializableTimestampAssigner<JSONObject>() {
                    @Override
                    public long extractTimestamp(JSONObject element, long recordTimestamp) {
                        return element.getLong("ts") * 1000L;
                    }
                }));

        // map
        SingleOutputStreamOperator<TestCourseUcScoreDuration> mapBeanDS = withWatermarkDS.map(jsonObj -> {
            String courseId = jsonObj.getString("course_id");
            String userId = jsonObj.getString("user_id");
            Long durationSec = jsonObj.getLong("duration_sec");
            Double score = jsonObj.getDouble("score");
            return new TestCourseUcScoreDuration("", "", courseId, "", 1, durationSec, score, 0, 0L, userId);
        });

        // 分组
        KeyedStream<TestCourseUcScoreDuration, String> keyedDS =
            mapBeanDS.keyBy(TestCourseUcScoreDuration::getCourseId);

        // process
        SingleOutputStreamOperator<TestCourseUcScoreDuration> processDS =
            keyedDS.process(new KeyedProcessFunction<String, TestCourseUcScoreDuration, TestCourseUcScoreDuration>() {
                // 状态编程获取独立用户
                private MapState<String, Integer> mapState;

                @Override
                public void open(Configuration parameters) throws Exception {
                    MapStateDescriptor<String, Integer> mapStateDescriptor =
                        new MapStateDescriptor<>("course-uv-state", String.class, Integer.class);
                    mapStateDescriptor.enableTimeToLive(StateTtlConfig.newBuilder(Time.days(1)).build());
                    mapState = getRuntimeContext().getMapState(mapStateDescriptor);
                }

                @Override
                public void processElement(TestCourseUcScoreDuration courseUcScoreDuration,
                    KeyedProcessFunction<String, TestCourseUcScoreDuration, TestCourseUcScoreDuration>.Context ctx,
                    Collector<TestCourseUcScoreDuration> out) throws Exception {

                    String userId = courseUcScoreDuration.getUserId();
                    if (!mapState.contains(userId)) {
                        courseUcScoreDuration.setUserCount(1);
                        mapState.put(userId, 1);
                    }
                    out.collect(courseUcScoreDuration);
                }
            });

        // 开窗
        AllWindowedStream<TestCourseUcScoreDuration, TimeWindow> windowDS = processDS
            .windowAll(TumblingEventTimeWindows.of(org.apache.flink.streaming.api.windowing.time.Time.seconds(10)));
        // 聚合
        SingleOutputStreamOperator<TestCourseUcScoreDuration> reduceDS =
            windowDS.reduce(new ReduceFunction<TestCourseUcScoreDuration>() {
                @Override
                public TestCourseUcScoreDuration reduce(TestCourseUcScoreDuration value1,
                    TestCourseUcScoreDuration value2) throws Exception {
                    value1.setTestCount(value1.getTestCount() + value2.getTestCount());
                    value1.setTotalDuration(value1.getTotalDuration() + value2.getTotalDuration());
                    value1.setTotalScore(value1.getTotalScore() + value2.getTotalScore());
                    value1.setUserCount(value1.getUserCount() + value2.getUserCount());
                    return value1;
                }
            }, new AllWindowFunction<TestCourseUcScoreDuration, TestCourseUcScoreDuration, TimeWindow>() {
                @Override
                public void apply(TimeWindow window, Iterable<TestCourseUcScoreDuration> values,
                    Collector<TestCourseUcScoreDuration> out) throws Exception {
                    String sst = DateFormatUtil.toYmdHms(window.getStart());
                    String edt = DateFormatUtil.toYmdHms(window.getEnd());
                    for (TestCourseUcScoreDuration ucScoreDuration : values) {
                        ucScoreDuration.setStt(sst);
                        ucScoreDuration.setEdt(edt);
                        ucScoreDuration.setTs(System.currentTimeMillis());
                        out.collect(ucScoreDuration);
                    }
                }
            });

        // 获取course_info中的courseName
        SingleOutputStreamOperator<TestCourseUcScoreDuration> courseDimDS =
            AsyncDataStream.unorderedWait(reduceDS, new DimAsyncFunction<TestCourseUcScoreDuration>("dim_course_info") {
                @Override
                public void join(TestCourseUcScoreDuration input, JSONObject dimInfo) {
                    input.setCourseName(dimInfo.getString("COURSE_NAME"));
                }

                @Override
                public String getKey(TestCourseUcScoreDuration obj) {
                    return obj.getCourseId();
                }
            }, 120, TimeUnit.SECONDS);

        courseDimDS.print("courseNameDS>>");

        // 写入
        courseDimDS.addSink(MyClickhouseUtil.<TestCourseUcScoreDuration>getSinkFunction(
            "insert into dws_test_course_uc_score_duration_window values(?,?,?,?,?,?,?,?,?)"));

        //
        env.execute();
    }
}